package com.gedi.data.dataclean;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Locale;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * @Author:RenPu
 * @Date: 2020/1/19 15:25
 * @Version: 1.0
 * @description:
 */


/**
 * 数据清洗的mapper
 */
public class DataCleanMapper extends Mapper<LongWritable, Text,Text, NullWritable> {

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        //设置正则表达式抽取数据的规则
        final String regex = "^(\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}\\.\\d{1,3}).*\\[(.*)\\]\\s\"\\w*\\s(.*)\\sHTTP\\/1.1\"\\s(\\d{3}).*$";

        //value代表从本地log文件读取的一行数据
        final String line = value.toString();

        final Pattern pattern = Pattern.compile(regex, Pattern.MULTILINE);
        final Matcher matcher = pattern.matcher(line);

        while (matcher.find()) {

            //请求客户端的Ip值
            String clientIp = matcher.group(1);

            //请求的时间
            String requestTime = matcher.group(2);

            //请求的资源
            String resource = matcher.group(3);

            //响应状态码
            String responseCode = matcher.group(4);


            /*
            时间格式的处理  时间格式设置与 30/May/2013:17:38:20 +0800 相对应
             30代表天，May代表月份（注意英文的年份需要MMM三个进行转换），2013代表年所以设置为（"dd/MMM/yyyy"）
             Locale.ENGLISH,国际化的时区
             */
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("dd/MMM/yyyy", Locale.ENGLISH);
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");

            try {
                Date parse = simpleDateFormat.parse(requestTime);
                String recorderTime = sdf.format(parse);
                context.write(new Text(clientIp+"--"+recorderTime+"--"+resource+"--"+responseCode),null);
            } catch (ParseException e) {
                e.printStackTrace();
            }




        }

    }
}
